package com.brid.KafkaProducer;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @Author: 司军强
 * @Date: 17:50 2019/12/19
 * @Description: 使用自定的的值序化器
 */
@Slf4j
public class ProducerDefineSerializer {

    //kfka对外提供的集群地址
    private static final String BROKERLIST = "192.168.223.128:9092";
    //topic 的name
    private static final String TOPIC = "hema";

    public static void main(String[] args) {
        //添加一些固定的属性值
        Properties prop = new Properties();
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, CompanySerializer.class.getName());
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CompanySerializer.class.getName());
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKERLIST);
        //创建kafka对象
        KafkaProducer producer = new KafkaProducer<>(prop);
        //构建对象值
        Company company = Company.builder().name("kafka").address("上海").build();
        //准备生产消息 指定topic 和值
        ProducerRecord record = new ProducerRecord<>(TOPIC, company);
        //发送消息
        try {
            Object o = producer.send(record).get();
            log.info("【push message success】 message-{}", JSONObject.toJSONString(o));
        } catch (Exception e) {
          log.error("【push message failure】 failure-{}",e);
        }
    }
}
